package cn.com.tt.flink.sql.engine.processor;

import cn.com.tt.flink.runner.AbstractProcessor;
import cn.com.tt.flink.sql.engine.pojo.JobConfig;
import cn.com.tt.flink.sql.engine.pojo.source.JobSource;
import cn.com.tt.flink.sql.engine.strategy.*;
import cn.com.tt.flink.sql.engine.utils.LambdaUtils;
import cn.com.tt.flink.sql.engine.strategy.*;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class RegisterSourceProcessor extends AbstractProcessor {

    public RegisterSourceProcessor(StreamTableEnvironment tableEnv) {
        super(tableEnv);
    }

    @Override
    public void process() throws Exception {
        JobConfig.getInstance().getSourceList().forEach(LambdaUtils.consumerWrapper(this::createTable));
    }

    public void createTable(JobSource source) throws Exception {
        String creatTableStatement = null;

        try {
            StrategyContext context;
            String connectorType = source.getConnectorType();

            switch (connectorType) {
                case "upsert-kafka": {
                    context = new StrategyContext(new UpsertKafkaTableStrategy(source));
                }
                break;
                case "hbase-1.4": {
                    context = new StrategyContext(new HbaseTableStrategy(source));
                }
                break;
                case "mysql-cdc": {
                    context = new StrategyContext(new MysqlCdcTableStrategy(source));
                }
                break;
                default: {
                    context = new StrategyContext(new KafkaTableStrategy(source));
                }
                break;
            }

            creatTableStatement = context.createTableStatement();
            tableEnv.executeSql(creatTableStatement);
        } catch (Exception e) {
            throw new Exception("脚本出错：" + creatTableStatement, e);
        }
    }

}
